package com.central.controller

import java.util

import com.alibaba.fastjson.JSON
import com.central.controller.ClueController.JobData
import com.central.service.TrailService
import com.central.utils.LoadESData
import com.typesafe.config.Config
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.scalactic.Accumulation.withGood
import org.scalactic._
import spark.jobserver.api.{JobEnvironment, SingleProblem, SparkJob, ValidationProblem}

import scala.util.parsing.json.JSONObject
import scala.util.{Failure, Success, Try}

object TrailController extends SparkJob {
  override type JobData = (SparkSession, TrailConditionEntity)
  override type JobOutput = Unit

  override def runJob(sc: SparkContext, runtime: JobEnvironment, data: JobData): JobOutput = {
    new TrailService().trailAnalysis(data._1, data._2)
  }

  override def validate(sc: SparkContext, runtime: JobEnvironment, config: Config): JobData Or Every[ValidationProblem] = {
    val properties = LoadESData.loadProperties();
    val spark = SparkSession
      .builder()
      .master("spark://192.168.5.180:7077")
      .config("es.nodes", properties.getProperty("elasticsearch.host", "192.168.5.180"))
      .config("es.port", properties.getProperty("elasticsearch.port", "9200"))
      .config("es.mapping.date.rich", "false")
      .getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")
    val parameterValidatin: JobData Or One[ValidationProblem] =
      Try { //将传入的参数转换成InitClueConditionEntity
        /**
          * json格式:{"jobId":"123", "relationship":"and/or/not",
          * "conditions":[
          * {\"deviceids\":\"195\",\"continue\":5,\"begintime\":\"2019-12-01 00:00:00\"},
          * {\"deviceids\":\"202\",\"continue\":5,\"begintime\":\"2019-12-01 00:00:00\"},
          * {\"deviceids\":\"196\",\"continue\":5,\"begintime\":\"2019-12-01 00:00:00\"}
          * ]}
          */
        import scala.collection.JavaConversions._
        val content = config.getConfig("body")
        val jobId = content.getInt("jobId")
        val threshold = content.getInt("threshold")

        val hasliferow = content.getString("hasliferow")
        val fuzzyway = content.getString("fuzzyway")

        import scala.collection.JavaConversions._

        val activityTimes = content.getAnyRefList("activitytimes").toList.map(jobj => {
          val timeMap = jobj.asInstanceOf[util.Map[String, Object]]
          val day = String.valueOf(timeMap.get("days"))
          val time = String.valueOf(timeMap.get("time"))
          ActivityTime(day, time)
        })
        println(content + "===============content")
        println(content.getAnyRef("conditions") + "===================content")
        val value = content.getAnyRef("conditions")
        if (value.isInstanceOf[String]) {
          val jarray = JSON.parseArray(value.asInstanceOf[String])
          jarray.map(x => {
            x.asInstanceOf[util.Map[String, Object]]
          })
        }
        val conditions = content
          .getAnyRefList("conditions")
          .toList
          .map(conditionNode => {
            val map = conditionNode.asInstanceOf[util.Map[String, Object]]
            val deviceids = String.valueOf(map.get("deviceids"))
            val deviceIdList = deviceids.split("_").toList
            val continue: Integer = java.lang.Integer.parseInt(map.get("continue") + "")
            val begintime = String.valueOf(map.get("begintime"))
            OnePointTrailConditionEntity(deviceIdList, continue, begintime)
          })
        TrailConditionEntity(jobId, threshold, conditions, activityTimes, hasliferow, fuzzyway)
      } match {
        case Success(context) => {
          Good((spark, context))
        }
        case Failure(ex) => Bad(One(SingleProblem("body is not exists or json can't parse. error message:" + ex.getMessage)))
      }
    withGood(parameterValidatin) { a => a }

  }
}

case class TrailConditionEntity(jobId: Integer,
                                threshold: Integer,
                                conditions: List[OnePointTrailConditionEntity],
                                activity: List[ActivityTime],
                                hasliferow: String,
                                fuzzyway: String
                               )

case class OnePointTrailConditionEntity(
                                         deviceids: List[String],
                                         continue: Integer,
                                         begintime: String
                                       )

//case class ActivityTime(
//                         days: String,
//                         time: String
//                       )